package com.example.mq.mqsever;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: WHY
 * Date: 2023-09-05
 * Time: 8:50
 */

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.mqsever.core.*;
import com.example.mq.mqsever.datacenter.DiskDataCenter;
import com.example.mq.mqsever.datacenter.MemoryDataCenter;
import javafx.beans.binding.MapExpression;
import org.apache.ibatis.javassist.bytecode.ExceptionsAttribute;
import org.springframework.web.bind.annotation.ExceptionHandler;

import java.io.DataInputStream;
import java.io.IOException;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 通过这个类,表示虚拟主机
 * 每个虚拟主机下面都管理着自己的交换机队列,绑定,消息,数据
 * 同时通过VirtualHost这个类作为业务逻辑的整合者,就需要对代码抛出的异常进行处理了
 */
public class VirtualHost {
    public String virtualHostName;
    public MemoryDataCenter memoryDataCenter=new MemoryDataCenter();
    public DiskDataCenter diskDataCenter=new DiskDataCenter();
    public Router router=new Router();

    public ConsumerManager consumerManager=new ConsumerManager(this);
    //针对交换机操作的锁对象
    public  final Object exchangeLocker=new Object();
    //针对队列的锁对象
    public final Object queueLocker=new Object();

    public String getVirtualHost() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public VirtualHost(String name){
        this.virtualHostName=name;


        //对于MemoryDataCenter来说,不需要进行额外的初始化操作,只要new出对象就行了
        //但是,针对DiskDataCenter来说,需要进行初始化操作,因为涉及到建库建表操作
        diskDataCenter.init();

        //另外还需要针对硬盘的数据进行恢复到内存的操作

        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException |MqException| ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost]恢复内存数据失败");
        }
    }
    //创建交换机
    //如果交换机不存在,就创建,如果存在就直接返回
    //返回值是boolean类型,创建成功返回TRUE,失败返回false
    public boolean exchangeDeclare(String exchangeName, Exchangetype exchangetype,boolean durable,boolean autoDelete,
     Map<String,Object>   arguments){

        //把交换机的名字加上虚拟机作为前缀,这样就可以允许交换机的名字重复了
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker) {
                //1.判定该交互机是否存在,直接通过内存查询
                Exchange existsExchange=   memoryDataCenter.getExchange(exchangeName);
                if(existsExchange!=null){
                    //该交换机已经存在
                    System.out.println("[VirtualHost] 交换机已经存在 exchangeName"+exchangeName);
                    return true;
                }
                //2.不存在,就创建
                Exchange exchange=new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangetype);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                //3.把构造好的交换机写入硬盘
                if(durable){
                    diskDataCenter.insertExchange(exchange);
                }
                //4.把交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost]交换机创建完成! exchangeName="+exchangeName);

                //上述逻辑先写硬盘,在写内存,目的就是硬盘更容易写失败,如果硬盘失败了,内存就不写了
                //要是先写内存,内存写成功了,硬盘写失败了,还需要把内存的数据给删掉,就比较麻烦了
            }

             return true;
        }catch(Exception e){
            System.out.println("[VirtualHost] 交换机创建失败! exchangeName="+exchangeName);
            e.printStackTrace();
            return false;
        }
    }


    //删除交换机
    public boolean exchangeDelete(String exchangeName){
        exchangeName=virtualHostName+exchangeName;
        try{
            synchronized (exchangeLocker){
                //1.先找到对应的交换机
                Exchange toDelete=memoryDataCenter.getExchange(exchangeName);
                if(toDelete==null){
                    throw new MqException("[VirtualHost] 交换机不存在无法删除!");
                }

                //2.删除硬盘上的数据
                if(toDelete.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //3.删除内存中的数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功! exchaneName" +exchangeName);
            }


            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName"+exchangeName);
            e.printStackTrace();
            return  false;
        }
    }
//创建队列
    public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete,
                                Map<String,Object> arguments){
        //把队列名的名字给拼接上虚拟主机的名字
        queueName=virtualHostName+queueName;
        try{
            synchronized (queueLocker){
                //1.判定队列是否存在
                MSGQueue existsQueue= memoryDataCenter.getQueue(queueName);
                if(existsQueue!=null){
                    System.out.println("[VirtualHost] 队列已经存在 queueName="+queueName );
                    return true;
                }
                //创建队列对象
                MSGQueue queue=new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setAutoDelete(autoDelete);
                queue.setExclusive(exclusive);
                queue.setArguments(arguments);
                //3.写硬盘
                if(durable){
                    diskDataCenter.insertQueue(queue);
                }

                //4.写内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName="+queueName);
            }

            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败! queueName="+queueName);
            e.printStackTrace();
            return false;
        }
    }
//删除队列
    public boolean queueDelete(String queueName){
        queueName=virtualHostName+queueName;
        try{
            synchronized (queueLocker){
                //1.根据队列名字,查询当下的队列对象
                MSGQueue queue= memoryDataCenter.getQueue(queueName);
                if(queue==null){
                    throw new MqException("[VirtualHost] 队列不存在! 无法删除! queueName="+queueName);
                }
                //2.删除硬盘数据
                if(queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                //3.删除内存数据

                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功! queueName= "+queueName);
            }

            return  true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除队列失败! queueName="+queueName);
            e.printStackTrace();
            return  false;
        }
    }
    //创建绑定

    public boolean queueBind(String queueName,String exchangeName,String bindingKey){
        queueName=virtualHostName+queueName;
        exchangeName=virtualHostName+exchangeName;
        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    //1.判定当前的绑定是否已经存在了
                    Binding existsBinding=memoryDataCenter.getBinding(exchangeName,queueName);
                    if(existsBinding!=null){
                        throw new MqException("[VirtualHost] binding已经存在 ! queueName= "+queueName);
                    }
                    //2.验证bindingKey是否合法
                    if(!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey非法! bindingKey="+bindingKey);
                    }


                    //3.创建Binding对象
                    Binding binding=new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    //4.获取一下对应的交换机和队列,如果交换机或者队列不存在,这样的绑定也是无法创建的
                    MSGQueue queue= memoryDataCenter.getQueue(queueName);
                    if(queue==null){
                        throw  new MqException("[VirtualHost] 队列不存在! queueName="+queueName);
                    }
                    Exchange exchange= memoryDataCenter.getExchange(exchangeName);
                    if(exchange==null){
                        throw new MqException("[VirtualHost] 交换机不存在! exchangeName="+exchangeName);
                    }
                    //先写硬盘
                    if(queue.isDurable()&&exchange.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }
                    //写入内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功! exchangeName="+exchangeName
                            +", queueName="+queueName);
                }
            }

            return true;

        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定创建失败! exchangeName="+exchangeName
                    +", queueName="+queueName);
            e.printStackTrace();
            return false;
        }
    }
    //删除绑定
    public boolean queueUnbind(String queueName,String exchangeName){
        queueName=virtualHostName+queueName;
        exchangeName=virtualHostName+exchangeName;

        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    //1.获取binding看是否已经存在

                    Binding binding= memoryDataCenter.getBinding(exchangeName,queueName);
                    if(binding==null){
                        throw new MqException("[VirtualHost] 删除绑定失败! 绑定不存在! exchangeName="+exchangeName
                                +", queueName="+queueName);
                    }
//        //2.获取一下对应的队列和交换机是否存在
//            Exchange exchange= memoryDataCenter.getExchange(exchangeName);
//            if(exchange==null){
//                throw new MqException("[VirtualHost] 对应的交换机不存在! exchangeName="+exchangeName);
//            }
//            MSGQueue queue= memoryDataCenter.getQueue(queueName);
//            if(queue==null){
//                throw new MqException("[VirtualHost] 对应的队列不存在! queueName="+queueName);
//            }
//            //3.删除硬盘上的数据
//            if(queue.isDurable()&&exchange.isDurable()){
                    //删除硬盘数据,无论是否持久化都尝试从硬盘删一下,就算不存在也没有副作用+
                    diskDataCenter.deleteBinding(binding);

                    //4.删除内存中的数据
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功!");
                }
            }

            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除绑定失败! ");
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到指定交换机/队列中
    public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body){
        try{
            //转换交换机名字
            exchangeName=virtualHostName+exchangeName;
            //2.检查routingKey的合法性
            if (!router.checkRoutingKey(routingKey)) {
                throw  new MqException("[VirtualHost] routingKey 非法! routingKey="+routingKey);


            }
            //3.查找交换机对象
            Exchange exchange= memoryDataCenter.getExchange(exchangeName);
            if(exchange==null){
                throw new MqException("[VirtualHost] 交换机不存在! exchangeName="+exchangeName);
            }
            //4.判定交换机类型
            if(exchange.getType()==Exchangetype.DIRECT){
                //按照直接交换机
                //以routingKey作为队列名字,直接把消息发送到该队列中
                //获取队列名
                String queueName=virtualHostName+routingKey;
                //构造一个消息对象
                Message message=Message.createMessageWithId(routingKey,basicProperties,body);
                //查找队列对应的对象

                MSGQueue queue= memoryDataCenter.getQueue(queueName);
                if(queue==null){
                    throw new MqException("[VirtualHost] 队列不存在! queueName="+queueName);
                }
                //队存在,给队列中写入消息
                sendMessage(queue,message);
            }else {
                //按照主题/扇出交换机
                //找到该交换机关联的所有绑定,并遍历这些绑定对象
                ConcurrentHashMap<String ,Binding> bindingsMap=memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String,Binding> entry:bindingsMap.entrySet()){
                    //1.获取到绑定对象,判定对应的队列是否存在
                    Binding binding=entry.getValue();
                    MSGQueue queue= memoryDataCenter.getQueue(binding.getQueueName());

                    if(queue==null){
                        //此处不抛异常了,可能这里有很多的队列
                        //不能因为一个队列是空的,影响到其他队列
                        System.out.println("[VirtualHost] basicPublish 发送消息时,发现队列不存在! queueName="+binding.getQueueName());
                        continue;
                    }
                    //2.构造消息对象
                    Message message=Message.createMessageWithId(routingKey,basicProperties,body);
                    //3.判定这个消息是否转发给队列,
                    //如果是fanout,所有绑定的队列都要转发
                    //如果是topic,还需要判定下,bindingKey和routingKey是不是匹配
                    if(!router.route(exchange.getType(),binding,message)){
                        continue;
                    }
                    //4.真正转发消息给队列
                    sendMessage(queue,message);
                }
            }

            return true;
        }catch(Exception e){
            System.out.println("[VirtualHost] 消息发送失败!");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        //发送消息,就是把消息写入内存和硬盘

        int deliveryMode=message.getDeliverMode();
        //delivery为1,不持久化,为2,持久化
        if(deliveryMode==2){
            diskDataCenter.sendMessage(queue,message);
        }
        //写入内存

        memoryDataCenter.sendMessage(queue,message);

        // 通知消费者可以消费消息了
            consumerManager.notifyConsume(queue.getName());
    }


    //订阅消息
    //添加一个队列的订阅者.当队列收到消息之后,就把消息推送给对应的订阅者
    //consumerTag:消费者的身份标识
    //autoAck: 消息被消费完成以后,应答的方式,为true,自动应答,为false,手动应答
    //consumer: 是一个回调函数,此处类型设为函数式接口,这样后续调用basicConsume并且传形参的时候,可以写作lambda的样子
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {

        //构造一个ConsumerEnv对象,把这个对应的队列找到,再把这个Consumer对象添加到该队列中
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume成功! queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
        public boolean basicAck(String queueName,String messageId) throws ClassNotFoundException {
        queueName=virtualHostName+queueName;

        try{
            //1.获取到消息和队列
            Message message=memoryDataCenter.getMessage(messageId);
            if(message==null){
        throw new MqException("[VirtualHost] 要确认的消息不存在! messageId="+messageId);
            }
            MSGQueue queue= memoryDataCenter.getQueue(queueName);
            //2.删除硬盘上的数据
            if(message.getDeliverMode()==2){
                diskDataCenter.deleteMessage(queue,message);
            }
            //3.删除消息中心中的数据
            memoryDataCenter.removeMessage(messageId);
            //4.删除待确认的集合中的数据
            memoryDataCenter.removeMessageWaitACK(queueName,messageId);
            System.out.println("[VirtualHost] basicAck成功! 消息被成功确认! queueName="+queueName
            +", messageId="+messageId);
            return  true;
        } catch (MqException | IOException e) {
            System.out.println("[VirtualHost] basicAck失败! 消息确认失败! queueName="+queueName
                    +", messageId="+messageId);
            e.printStackTrace();
            return false;
        }

        }
}
